package com.geek.bulk.interceptor;

import org.eclipse.microprofile.faulttolerance.Asynchronous;
import org.eclipse.microprofile.faulttolerance.Bulkhead;

import javax.interceptor.Interceptor;
import javax.interceptor.InvocationContext;
import java.lang.annotation.Annotation;
import java.lang.reflect.Method;
import java.util.Objects;
import java.util.concurrent.*;


/**
 * 舱壁隔离(Bulkhead Isolation), 借用船舶行业设计理念来理解
 */
@Interceptor
public class BulkInterceptor extends AnnotatedInterceptor<Bulkhead>{

    private ConcurrentHashMap<Bulkhead, ExecutorService> executorsCache = new ConcurrentHashMap<>();

    private ConcurrentHashMap<Bulkhead, Semaphore> semaphoreCache = new ConcurrentHashMap<>();


    @Override
    protected Object execute(InvocationContext context, Bulkhead bulkhead) throws Exception {
        Method method = context.getMethod();
        //判断 走信号量模式还是 计数模式
        if (isThreadIsolation(method)) {
            //计数
            return executeInThreadIsolation(context, bulkhead);
        } else {
            //信号量
            return executeInSemaphoreIsolation(context, bulkhead);
        }
    }

    /**
     * 控制并发 (Thread Isolation 线程隔离)
     */
    private Object executeInThreadIsolation(InvocationContext context, Bulkhead bulkHead) throws ExecutionException, InterruptedException {
        ExecutorService service = executorsCache.computeIfAbsent(bulkHead, key -> {
            int fixSize = key.value();
            int waitingTaskQueue = bulkHead.waitingTaskQueue();
            ExecutorService executorService = new ThreadPoolExecutor(fixSize, fixSize, 0, TimeUnit.MILLISECONDS, new ArrayBlockingQueue<>(waitingTaskQueue));
            return executorService;
        });
        Future<Object> submit = service.submit(context::proceed);
        return submit.get();
    }

    /**
     * 控制并发 (Semaphore Isolation 信号量隔离)
     * 属于一种较常见的限流手段，在工程实践中可以通过信号量机制（如Java中的Semaphore）来控制
     */
    private Object executeInSemaphoreIsolation(InvocationContext context, Bulkhead bulkHead) throws Exception {
        Semaphore semaphore = semaphoreCache.computeIfAbsent(bulkHead, key -> {
            int value = key.value();
            return  new Semaphore(value);
        });
        try {
            semaphore.acquire();
            return context.proceed();
        }finally {
            semaphore.release();
        }
    }


    private Bulkhead findBulkHead(Method method) {
        for (Annotation annotation : method.getDeclaredAnnotations()) {
            if (Objects.equals(annotation.annotationType(), Bulkhead.class)) {
                return (Bulkhead)annotation;
            }
        }
        return null;
    }

    private boolean isThreadIsolation(Method method) {
        return method.isAnnotationPresent(Asynchronous.class);
    }

}
